package com.dctp.common.disruptor.close;

import com.dctp.common.cache.ProfitStatCache;
import com.dctp.common.service.TradeService;
import com.dctp.common.vo.ProfitStat;
import com.jfinal.plugin.activerecord.Db;
import com.jfinal.plugin.activerecord.Record;
import com.lmax.disruptor.EventHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

import static com.dctp.common.service.TradeService.publishCloseOrder;

public class CloseOrderEventHandler implements EventHandler<CloseOrderEvent> {

    private static final Logger logger = LoggerFactory.getLogger(CloseOrderEventHandler.class);

    // ✅ 批次缓存
    private final List<CloseOrder> batchBuffer = new ArrayList<>(200);
    private final int batchSize = 50;

    // ✅ 异步线程池（用于缓存更新/发布事件）
    private final ExecutorService asyncExecutor = Executors.newFixedThreadPool(4);

    @Override
    public void onEvent(CloseOrderEvent event, long sequence, boolean endOfBatch) {
        if (event.getOrder() != null) {
            batchBuffer.add(event.getOrder());
        }
        if (batchBuffer.size() >= batchSize || endOfBatch) {
            flushBatch();
        }
    }

    /** ✅ flush 批次 */
    private void flushBatch() {
        try {
            // ✅ Step1: 写库（事务保证）
            boolean success = false;
            // ✅ 最多重试 3 次
            for (int i = 0; i < 3; i++) {
                try {
                    Db.tx(() -> {
                        if (batchBuffer.size() == 1) {
                            writeSingle(batchBuffer.get(0));
                        } else {
                            writeBatch(batchBuffer);
                        }
                        logger.info("平仓批处理订单数：{}",batchBuffer.size());
                        return true;
                    });
                    success = true;
                    break;  // ✅ 成功就跳出循环
                } catch (Exception e) {
                    logger.error("⚠️ 第 {} 次批量写库失败，重试中...", i + 1, e);
                    Thread.sleep(200); // ✅ 给 DB 缓一缓（200ms）
                }
            }

            // ❌ 如果 3 次都失败 → 落地文件
            if (!success) {
                saveToFallbackFile(batchBuffer);
                return;  // ⚠️ 提前结束（不执行缓存更新/发布逻辑）
            }

            // ✅ Step2: 查询当前批次用户持仓（只查相关用户，不扫全表）
            Set<Long> userIds = batchBuffer.stream()
                    .map(o -> o.getUpdateUser().getLong("id"))
                    .collect(Collectors.toSet());

            String placeholders = userIds.stream()
                    .map(id -> "?")
                    .collect(Collectors.joining(","));

            String sql = "SELECT user_id, symbol, COUNT(*) AS cnt " +
                    "FROM dctp_order " +
                    "WHERE status = 1 AND user_id IN (" + placeholders + ") " +
                    "GROUP BY user_id, symbol";

            List<Record> holdings = Db.find(sql, userIds.toArray());

            // ✅ Step3: 转换成 Map 结构
            Map<Long, Map<String, Integer>> holdingMap = new HashMap<>();
            for (Record r : holdings) {
                holdingMap
                        .computeIfAbsent(r.getLong("user_id"), k -> new HashMap<>())
                        .put(r.getStr("symbol"), r.getInt("cnt"));
            }

            //线程池异步执行
            asyncExecutor.submit(()-> {
                TradeService.afterOrderClosed(batchBuffer);
            });

            // ✅ Step4: 异步执行缓存更新 & publishCloseOrder
            for (CloseOrder order : batchBuffer) {
                asyncExecutor.submit(() -> processAfterClose(order, holdingMap));
            }

        } catch (Exception e) {
            logger.error("❌ 批量写订单失败", e);
            saveToFallbackFile(batchBuffer);
        } finally {
            batchBuffer.clear();
        }
    }

    /** ✅ 单条订单写库 */
    private void writeSingle(CloseOrder order) {
        // 更新用户余额
        Db.update("UPDATE dctp_user SET balance = GREATEST(balance + ?, 0) WHERE id = ?",
                order.getUpdateUser().getBigDecimal("refund"),
                order.getUpdateUser().getLong("id"));

        // 更新订单
        Db.update("dctp_order", order.getOrder());

        // 插入资金流水
        Db.save("dctp_fund_flow", order.getServiceFeeFundFlow());
        if (order.getOrderFundFlow() != null) {
            Db.save("dctp_fund_flow", order.getOrderFundFlow());
        }
    }

    /** ✅ 批量订单写库 */
    private void writeBatch(List<CloseOrder> orders) {
        List<Record> orderUpdates = new ArrayList<>();
        List<Record> fundFlows = new ArrayList<>();
        List<Record> userUpdates = new ArrayList<>();

        for (CloseOrder order : orders) {
            orderUpdates.add(order.getOrder());
            userUpdates.add(order.getUpdateUser());
            fundFlows.add(order.getServiceFeeFundFlow());
            if (order.getOrderFundFlow() != null) {
                fundFlows.add(order.getOrderFundFlow());
            }
        }

        // 准备批量更新参数
        Object[][] userParams = new Object[userUpdates.size()][2];
        for (int i = 0; i < userUpdates.size(); i++) {
            Record r = userUpdates.get(i);
            userParams[i][0] = r.getBigDecimal("refund");
            userParams[i][1] = r.getLong("id");
        }

        // 批量更新用户余额
        Db.batch("UPDATE dctp_user SET balance = GREATEST(balance + ?, 0) WHERE id = ?",
                userParams, userUpdates.size());

        // 批量更新订单
        Db.batchUpdate("dctp_order", orderUpdates, orderUpdates.size());

        // 批量插入资金流水
        Db.batchSave("dctp_fund_flow", fundFlows, fundFlows.size());
    }

    /** ✅ 异步处理缓存更新 & publishCloseOrder */
    private void processAfterClose(CloseOrder order, Map<Long, Map<String, Integer>> holdingMap) {
        Long userId = order.getUpdateUser().getLong("id");
        String symbol = order.getOrder().getStr("symbol");

        int totalCount = holdingMap.getOrDefault(userId, Map.of())
                .values().stream().mapToInt(Integer::intValue).sum();

        int symbolCount = holdingMap.getOrDefault(userId, Map.of())
                .getOrDefault(symbol, 0);

        // ✅ 更新 ProfitStatCache
        if (totalCount == 0) {
            ProfitStat profitStat = ProfitStatCache.get(userId);
            if (profitStat != null) {
                logger.info("当前用户：{}，没有持仓，清空浮动收益", userId);
                BigDecimal refund = order.getUpdateUser().getBigDecimal("refund");
                profitStat.setDynamicEquity(order.getUpdateUser().getBigDecimal("balance").add(refund));
                profitStat.setTotalPnl(BigDecimal.ZERO);
                ProfitStatCache.set(userId, profitStat);
            }
        }
        // ✅ 发布平仓事件
        if (totalCount == 0 || symbolCount == 0) {
            logger.info("用户：{},symbol：{}，持仓数为0",userId,symbol);
            publishCloseOrder(order.getOrder());
        }



    }

    /** ✅ 把失败的订单批次落地到文件 */
    private void saveToFallbackFile(List<CloseOrder> orders) {
        try {
            String dir = "/tmp/close_order_fallback"; // ⚠️ 可改为 NFS/持久目录
            java.nio.file.Files.createDirectories(java.nio.file.Paths.get(dir));

            String fileName = dir + "/batch_" + System.currentTimeMillis() + ".json";
            try (java.io.FileWriter fw = new java.io.FileWriter(fileName)) {
                fw.write(toJson(orders));
            }
            logger.error("🚨 已将 {} 条订单写入 fallback 文件: {}", orders.size(), fileName);
        } catch (Exception e) {
            logger.error("❌ fallback 文件写入失败（数据丢失风险）", e);
        }
    }

    /** ✅ 简单 JSON 序列化（可换 Jackson/Gson） */
    private String toJson(List<CloseOrder> orders) {
        StringBuilder sb = new StringBuilder("[");
        for (int i = 0; i < orders.size(); i++) {
            CloseOrder o = orders.get(i);
            sb.append(o.toJson()); // ⚠️ 需要 CloseOrder 有 toJson() 或者手动拼 JSON
            if (i < orders.size() - 1) sb.append(",");
        }
        sb.append("]");
        return sb.toString();
    }
}
